package com.de.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.de.controller.IndexController;
import com.de.entity.AjaxResult;
import com.de.entity.ImageEncoder;
import com.de.rtsp.MediaTransfer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


@ServerEndpoint(value = "/webSocketService/{userId}", encoders = {ImageEncoder.class})
@Component
@Slf4j
public class WebSocketServer {
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";

    private static final int MAX_CONNECTIONS = 100; // 最大连接数限制
    private static final Lock connectionLock = new ReentrantLock();

    /**
     * 连接建立成功调用的方法
     */
    @SneakyThrows
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        connectionLock.lock();
        try {
            if (webSocketMap.size() >= MAX_CONNECTIONS) {
                session.close();
                log.warn("达到最大连接数限制，拒绝新的连接请求");
                return;
            }

            this.session = session;
            // 设置较大的消息缓冲区大小
            session.setMaxBinaryMessageBufferSize(1024 * 1024);
            session.setMaxTextMessageBufferSize(1024 * 1024);

            if (webSocketMap.containsKey(userId)) {
                // 关闭旧连接
                WebSocketServer oldServer = webSocketMap.get(userId);
                try {
                    oldServer.session.close();
                } catch (IOException e) {
                    log.error("关闭旧连接失败", e);
                }
                webSocketMap.remove(userId);
            }

            webSocketMap.put(userId, this);
            addOnlineCount();

            log.info("用户连接:{},当前在线人数为:{}", userId, getOnlineCount());
            sendInitialMessage(userId);
        } finally {
            connectionLock.unlock();
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        try {
            if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
                webSocketMap.remove(userId);
                subOnlineCount();

                cleanupMediaTransfer(userId, "连接关闭");
            }
        } catch (Exception e) {
            log.error("关闭WebSocket连接时发生错误，userId: {}", userId, e);
        } finally {
            log.info("用户退出: {}，当前在线人数为: {}", userId, getOnlineCount());
        }
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息,必须是json串
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        if (StringUtils.isNotBlank(message)) {
            //解析发送的报文
            JSONObject jsonObject = JSON.parseObject(message);
            //追加发送人(防止串改)
            if (jsonObject != null) {
                jsonObject.put("fromUserId", this.userId);
                String toUserId = jsonObject.getString("toUserId");
                //传送给对应toUserId用户的websocket
                if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {
                    webSocketMap.get(toUserId).sendMessageByStr(jsonObject.toJSONString());
                } else {
                    log.error("请求的userId:" + toUserId + "不在该服务器上");
                    //否则不在这个服务器上，发送到mysql或者redis
                }
            }
        }
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        try {
            log.error("WebSocket发生错误，userId: {}，原因: {}", userId, error.getMessage());
            log.error("WebSocket详细错误: ", error);

            cleanupMediaTransfer(userId, "连接错误");
        } catch (Exception e) {
            log.error("处理WebSocket错误时发生异常，userId: {}", userId, e);
        }
    }

    public void sendMessageByStr(String message) {
        if (StringUtils.isNotBlank(message)) {
            try {
                this.session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.error("发送到用户：" + this.userId + "信息失败 ，信息是：" + message);
                log.error("websocket send str msg exception: ", e);
            }
        }
    }

    public void sendMessageByObject(Object message) {
        if (message != null) {
            try {
                this.session.getBasicRemote().sendObject(message);
            } catch (IOException | EncodeException e) {
                log.error("发送到用户：" + this.userId + "信息失败 ，信息是：" + message);
                log.error("websocket send object msg exception: ", e);
            }
        }
    }

    public void sendBinary(ByteBuffer message) {
        if (message != null) {
            webSocketMap.get(userId).sendBinary(message);
        }
    }

    /**
     * 发送自定义消息
     */
    public static void sendInfo(String message, String userId) {
        if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessageByStr(message);
        } else {
            log.error("用户" + userId + ",不在线！");
        }
    }

    public static void sendInfo(Object message, String userId) {
        if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessageByObject(message);
        } else {
            log.error("用户" + userId + ",不在线！");
        }
    }

    /**
     * 向所有的客户端发送消息
     *
     * @param byteBuffer byteBuffer
     * @throws IOException IOException
     */
    public static void sendAllByBinary(ByteBuffer byteBuffer) {
        if (!webSocketMap.isEmpty()) {
            Collection<WebSocketServer> values = webSocketMap.values();
            for (WebSocketServer next : values) {
                next.sendBinary(byteBuffer);
            }
        }
    }

    public static void sendAllByObject(Object message) {
        if (!webSocketMap.isEmpty()) {
            Collection<WebSocketServer> values = webSocketMap.values();
            for (WebSocketServer next : values) {
                next.sendMessageByObject(message);
            }
        }

    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }

    /**
     * 清理指定用户的媒体传输资源
     *
     * @param userId 用户ID
     * @param reason 清理原因
     */
    private void cleanupMediaTransfer(String userId, String reason) {
        if (StringUtils.isBlank(userId)) {
            return;
        }

        MediaTransfer mediaTransfer = IndexController.mediaTransfers.get(userId);
        if (mediaTransfer != null) {
            try {
                mediaTransfer.stopStreaming();
                IndexController.mediaTransfers.remove(userId);
                log.info("已清理用户 {} 的推流器，原因: {}", userId, reason);
            } catch (Exception e) {
                log.error("清理用户 {} 的推流器时发生错误", userId, e);
            }
        }
    }

    /**
     * 发送初始连接消息
     * @param userId 用户ID
     */
    private void sendInitialMessage(String userId) {
        try {
            JSONObject message = new JSONObject();
            message.put("type", "connection");
            message.put("userId", userId);
            message.put("status", "connected");
            message.put("timestamp", System.currentTimeMillis());

            sendMessageByStr(message.toJSONString());
        } catch (Exception e) {
            log.error("发送初始连接消息失败，userId: {}", userId, e);
        }
    }

}
